package com.xiguthings.xiniu.iot.device.state.kafka;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
public class DataHandlerPool {

	private final Logger LOGGER = LoggerFactory.getLogger(DataHandlerPool.class);

	// kafka客户端线程池的大小
	@Value("${kafka.consumer.pool.thread.count}")
	public int kafkaConsumerPoolThreadCount;

	/**
	 * 线程池
	 */
	private ExecutorService executor;

	/**
	 * 初始化
	 */
	@PostConstruct
	void init() {
		executor = Executors.newFixedThreadPool(kafkaConsumerPoolThreadCount);
	}

	/**
	 * 提交新的消费者
	 *
	 * @param Consumer
	 */
	public void submitConsumerPool(Runnable r) {
		executor.execute(r);
	}

	/**
	 * 获取已经激活的线程
	 */
	public int getActiveCount() {
		ThreadPoolExecutor pool = (ThreadPoolExecutor) executor;
		int count = pool.getActiveCount();
		return count;
	}

	/**
	 * 程序关闭,关闭线程池
	 */
	@PreDestroy
	void fin() {
		shutdown();
	}

	public void shutdown() {
		if (executor != null)
			executor.shutdown();
		try {
			if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
				LOGGER.info("Timed out waiting for consumer threads to shut down, exiting uncleanly");
			}
		} catch (InterruptedException e) {
			LOGGER.info("Interrupted during shutdown, exiting uncleanly");
		}
	}
}
